package com.joseph.websocket.handler;

import com.alibaba.fastjson.JSON;
import com.joseph.websocket.bean.Message;
import com.joseph.websocket.bean.SessionUser;
import com.joseph.websocket.bean.UserInfo;
import com.joseph.websocket.component.SessionManager;
import com.joseph.websocket.test.A;
import com.joseph.websocket.util.AttrUtil;
import com.joseph.websocket.util.CmdEnum;
import com.joseph.websocket.util.JedisUtil;
import com.joseph.websocket.util.MessageUtil;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.CharsetUtil;
import io.netty.util.HashedWheelTimer;
import io.netty.util.concurrent.*;
import lombok.extern.log4j.Log4j;
import lombok.extern.log4j.Log4j2;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.dubbo.qos.command.annotation.Cmd;
import org.omg.CORBA.CTX_RESTRICT_SCOPE;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

@ChannelHandler.Sharable
@Log4j2
public class WebSocketHandler extends SimpleChannelInboundHandler<Object> {

    protected void channelRead0(ChannelHandlerContext ctx, Object frame) throws Exception {
        //纯文本
        if (frame instanceof TextWebSocketFrame) {
            handleTextProtocol((TextWebSocketFrame) frame, ctx);
        //二进制格式
        } else if (frame instanceof BinaryWebSocketFrame) {
            handleBinProtocol((BinaryWebSocketFrame) frame, ctx);
        }
    }

    private void handleBinProtocol(BinaryWebSocketFrame frame, ChannelHandlerContext ctx) {
        ByteBuf content = frame.content();
        final int length = content.readableBytes();
        final byte[] array = new byte[length];
        content.getBytes(content.readerIndex(), array, 0, length);
        String msg = new String(array);
        Message message = JSON.parseObject(msg, Message.class);
        if (message.getCmd() == CmdEnum.MESSAGE.getCode()) {

        } else if (message.getCmd() == CmdEnum.BIZ.getCode()) {
            handleBiz(ctx, message);
        }
    }

    private void handleBiz(ChannelHandlerContext ctx, Message message) {
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread());
        Message resp = new Message();
        resp.setContent("biz");
        ByteBuf byteBuf = Unpooled.copiedBuffer(JSON.toJSONString(resp), CharsetUtil.UTF_8);
        ctx.writeAndFlush(new BinaryWebSocketFrame(byteBuf));
    }

    private void handleTextProtocol(TextWebSocketFrame frame, ChannelHandlerContext ctx) {
        long now = System.currentTimeMillis();
        String msg = frame.text();
        if (StringUtils.isBlank(msg)) {
            return;
        }
        Channel channel = ctx.channel();
        Message message;
        try {
            message = JSON.parseObject(msg, Message.class);
            //content内容加密
            if (message.getCryptoType() != null) {

            }
            log.info("recv msg from client: " + msg);
        } catch (Exception e) {
            log.error("消息解析异常，{}", msg, e);
            return;
        }

        if (message.getCmd() == null) {
            return;
        }

        //处理心跳包
        if (message.getCmd() == CmdEnum.HEART.getCode()) {
            Message heart = new Message();
            heart.setCmd(CmdEnum.HEART.getCode());
            MessageUtil.send(ctx, heart);
            Message response = new Message();
            response.setContent("aaaaaas");
            response.setCmd(CmdEnum.MESSAGE.getCode());
            response.setToId("1");
            response.setType(0);
            MessageUtil.send(ctx, response);
            return;
        }

        Long lastMsgTime = channel.attr(AttrUtil.USER_SEND_LAST_MSG_TIME).get();
        //这是限制发送消息频率，只记录用户主动发送的消息时间，比较last time
        if (message.getCmd() == CmdEnum.MESSAGE.getCode()) {
            channel.attr(AttrUtil.USER_SEND_LAST_MSG_TIME).set(now);
//                if (lastMsgTime != null && now - lastMsgTime <= 1000) {
//                    Message limitMessage = new Message();
//                    limitMessage.setSeqId(message.getSeqId());
//                    limitMessage.setCmd(CmdEnum.LIMIT.getCode());
//                    MessageUtil.send(ctx, limitMessage);
//                    return;
//                }
        }

        String msgId = UUID.randomUUID().toString();
        //JedisUtil.getJedis().set(RedisKeyUtil.SERVER_UN_CHECK_MESSAGE + msgId, msg);
        String userStr = JedisUtil.getJedis().get(message.getFromId());
        UserInfo userInfo = JSON.parseObject(userStr, UserInfo.class);
        if (userInfo == null || StringUtils.isEmpty(userInfo.getSessions().get(message.getStoreCode()).getSessionId())) {
            return;
        }
        message.setMsgTime(now);
        message.setMsgId(msgId);
        message.setSessionId(userInfo.getSessions().get(message.getStoreCode()).getSessionId());
        //丢MQ消息落地

        //返回客服端消息确认包
        Message ackMessage = new Message();
        ackMessage.setCmd(CmdEnum.ACK.getCode());
        ackMessage.setSeqId(message.getSeqId());
        ackMessage.setMsgTime(now);
        ackMessage.setMsgId(msgId);
        MessageUtil.send(ctx, ackMessage);

        SessionUser toUser = SessionManager.getUser(message.getToId());
        if (toUser != null) {
            toUser.send(message);
        } else {
            System.out.println("to user offline : " + message.getToId());
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("websockethandler inactive");
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
    }

    private void handleMsgCrypto(Message message) {
        if (message.getCryptoType() == 1) {
            //message.setContent();
        }
    }

    public static void main(String[] args) {
        HashedWheelTimer timer = new HashedWheelTimer();
    }
}